package com.fwmagic.flink.state;

import com.fwmagic.flink.state.source.MyExcetlyOnceParallFileSource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/***
 * 自定义并发读取文件的Source,且支持Exactly-Once
 * 不同subTask读取不同的文件
 * 准备数据：
 *      /Users/fangwei/temp/data/0.txt
 *      /Users/fangwei/temp/data/1.txt
 * 错误：Caused by: java.io.FileNotFoundException: /Users/fangwei/temp/data/2.txt (No such file or directory)
 * 原因：默认并行度根据电脑配置来，我这默认12个并行度
 * 解决：临时办法，将并行度设置为2
 */
public class OperatorStateDemo {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        env.enableCheckpointing(5000);

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000));

        env.setStateBackend(new FsStateBackend("file:///Users/fangwei/Downloads/chk"));

        //程序异常退出或者人工cancel掉，checkpoint不删除，保留！
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        DataStreamSource<String> dataStream = env.socketTextStream("localhost", 8888);

        //手动制造异常：用于测试checkpoint数据被正确存储和读取
        dataStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String line) throws Exception {
                if (line.equals("err")) {
                    throw new RuntimeException("程序异常！");
                }
                return line;
            }
        }).print();


        DataStreamSource<Tuple2<Integer, String>> data = env.addSource(new MyExcetlyOnceParallFileSource("/Users/fangwei/temp/data"));

        data.print();

        env.execute("MyExcetlyOnceParallFileSource");

    }
}
